import io.openmessaging.BytesMessage;
import io.openmessaging.demo.DefaultBytesMessage;
import io.openmessaging.demo.MessageUtils2;

import java.io.File;
import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;

/**
 * Author :  Rocky
 * Date : 24/05/2017 23:08
 * Description :
 * Test :
 */
public class MultiThreadMultiFilesWriteSingleMessageTest {

    public static void main(String[] args) throws InterruptedException {
        long s = System.currentTimeMillis();

        Semaphore semaphore = new Semaphore(2, true);

        BytesMessage message = buildMsg();

        for (int j = 0; j < 10; j++) {

            CountDownLatch latch = new CountDownLatch(10);
            for (int i = 0; i < 10; i++) {
                int finalI = i;
                Thread t = new Thread(new Runnable() {
                    @Override
                    public void run() {

                        try {
                            semaphore.acquire();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }

                        String fileName = "testfile" + finalI;
                        try {
                            File file = new File(fileName);
                            if (file.exists()) {
                                file.delete();
                            }
                            RandomAccessFile raf = new RandomAccessFile(file, "rw");
                            MappedByteBuffer mbb = raf.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, 200 * 1024 * 1024);

                            boolean writeSuccess = true;
                            while (mbb.hasRemaining() && writeSuccess) {
                                writeSuccess = MessageUtils2.writeMessage(message, mbb);
                            }

                        } catch (Exception e) {
                            e.printStackTrace();
                        }

                        semaphore.release();
                        latch.countDown();
                    }
                });
                t.start();
            }

            latch.await();
        }

        long time = (System.currentTimeMillis() - s);
        long totalBytes = 20000; //单位m
        System.out.println("10个线程10个文件，每个线程负责写一个文件，各写200m，循环10次， 总耗时 " + time + " ms ，速度 " + (totalBytes * 1024 / (time * 1.0 / 1000)) + " kb/s");
    }


    private static BytesMessage buildMsg(String topic) {
        BytesMessage message = buildMsg();
        message.headers().put("Topic", topic);
        return message;
    }

    private static BytesMessage buildMsg() {
        BytesMessage msg = new DefaultBytesMessage();
        msg.putHeaders("1", 1);
        msg.putHeaders("2", 2l);
        msg.putHeaders("3", 3.0);
        msg.putHeaders("4", "4");
        msg.putProperties("5", 5);
        msg.putProperties("6", 6l);
        msg.putProperties("7", 7.0);
        msg.putProperties("8", "8");
        msg.setBody("I LOVE YOU".getBytes());
        return msg;
    }
}
